package com.sunlands.deskmate.message.handler.sender.websocket;

import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.message.handler.sender.RetrySender;
import com.sunlands.deskmate.message.handler.sender.Sender;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import java.nio.channels.ClosedChannelException;

/**
 * @Author: anjunling@sunlands.com
 * @Description:
 * @Date: Create in 12:48 2018/6/14
 * @Modified By:
 */
@Slf4j
public class DefaultWebsocketMessageSender extends AbstractWebsocketMessageSender<Messages> implements Sender<Messages, Integer> {

    public DefaultWebsocketMessageSender(ChannelInfo channelInfo, Messages messages) {
        this.channelInfo = channelInfo;
        this.message = messages;
    }

    @Override
    protected ChannelFutureListener futureListener() {
        return new DefaultWebsocketMessageSender.SendChannelFutureListener(this);
    }

    @Override
    public void send() {
        super.send();
    }

    @Override
    public Integer strategy() {
        return 0;
    }

    private static class SendChannelFutureListener implements ChannelFutureListener {


        private DefaultWebsocketMessageSender defaultWebsocketMessageSender;

        public SendChannelFutureListener(DefaultWebsocketMessageSender defaultWebsocketMessageSender) {
            this.defaultWebsocketMessageSender = defaultWebsocketMessageSender;
        }

        @Override
        public void operationComplete(ChannelFuture channelFuture) {
            synchronized (defaultWebsocketMessageSender) {
                try {
                    if (!channelFuture.isSuccess()) {
                        Throwable throwable = channelFuture.cause();
                        if (throwable instanceof ClosedChannelException) {
                            defaultWebsocketMessageSender.releaseChannel(channelFuture.channel());

                            log.warn("send channel future close channelInfo {}", defaultWebsocketMessageSender.channelInfo);
                            defaultWebsocketMessageSender.isFinish = true;
                            defaultWebsocketMessageSender.isDone = true;
                        } else {
                            //指定异常情况处理
                            log.error("send channel future exception channelInfo " + defaultWebsocketMessageSender.channelInfo, throwable);

                            //转移到重试
                            if (defaultWebsocketMessageSender.strategy() <= 0) {
                                RetrySender<Messages, Integer> retrySender =
                                        new RetryWebsocketMessageSender(defaultWebsocketMessageSender.channelInfo, defaultWebsocketMessageSender.message);
                                retrySender.retry();
                            }

                            defaultWebsocketMessageSender.isDone = true;
                        }
                    } else {
                        defaultWebsocketMessageSender.isSuccess = true;
                        defaultWebsocketMessageSender.isFinish = true;
                        defaultWebsocketMessageSender.isDone = true;
                    }
                } catch (Exception e) {
                    log.error("send future listener channelInfo " + defaultWebsocketMessageSender.channelInfo, e);
                } finally {
                    if (log.isDebugEnabled()) {
                        log.debug("ws sender notify all");
                    }
                    defaultWebsocketMessageSender.notifyAll();
                }
            }
        }
    }
}
